package top.reminis.monitor;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.Gauge;
import org.apache.flink.metrics.MetricGroup;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 通用Flink作业监控系统
 * 提供实时的数据处理指标监控，包括吞吐量、延迟、错误率等
 * 支持任何类型的Flink作业监控
 * 
 * @author zwf
 * @create 2025-09-25
 */
@Slf4j
public class UniversalJobMonitor {
    
    private static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    
    // 作业信息
    private static volatile String jobName = "Unknown Job";
    private static volatile String jobDescription = "";
    private static volatile int monitoringInterval = 60; // 默认60秒
    
    // 计数器
    private static final AtomicLong totalRecordsProcessed = new AtomicLong(0);
    private static final AtomicLong totalRecordsSuccess = new AtomicLong(0);
    private static final AtomicLong totalRecordsError = new AtomicLong(0);
    private static final AtomicLong totalBytesProcessed = new AtomicLong(0);
    
    // 时间戳
    private static volatile LocalDateTime jobStartTime;
    private static volatile LocalDateTime jobEndTime;
    private static volatile LocalDateTime lastProcessTime;
    
    // Flink Metric相关
    private static Counter recordsProcessedCounter;
    private static Counter recordsSuccessCounter;
    private static Counter recordsErrorCounter;
    private static Counter bytesProcessedCounter;
    private static Gauge<Long> processingRateGauge;
    private static Gauge<Double> successRateGauge;
    private static Gauge<Double> errorRateGauge;
    
    /**
     * 初始化监控系统
     * 
     * @param jobName 作业名称
     * @param jobDescription 作业描述
     * @param monitoringInterval 监控间隔（秒）
     */
    public static void initialize(String jobName, String jobDescription, int monitoringInterval) {
        UniversalJobMonitor.jobName = jobName != null ? jobName : "Unknown Job";
        UniversalJobMonitor.jobDescription = jobDescription != null ? jobDescription : "";
        UniversalJobMonitor.monitoringInterval = monitoringInterval > 0 ? monitoringInterval : 60;
        
        jobStartTime = LocalDateTime.now();
        lastProcessTime = jobStartTime;
        
        log.info("=== {} 监控系统启动 ===", UniversalJobMonitor.jobName);
        log.info("作业描述: {}", UniversalJobMonitor.jobDescription);
        log.info("作业开始时间: {}", formatTime(jobStartTime));
        log.info("监控间隔: {} 秒", UniversalJobMonitor.monitoringInterval);
    }
    
    /**
     * 初始化Flink Metric监控
     * 
     * @param runtimeContext Flink运行时上下文
     */
    public static void initializeMetrics(RuntimeContext runtimeContext) {
        if (runtimeContext == null) {
            log.warn("RuntimeContext为空，跳过Flink Metric初始化");
            return;
        }
        
        try {
            log.info("初始化Flink Metric监控系统...");
            
            // 获取Metric组
            MetricGroup metricGroup = runtimeContext.getMetricGroup()
                    .addGroup("flink_job")
                    .addGroup("data_processing");
            
            // 创建计数器
            recordsProcessedCounter = metricGroup.counter("records_processed_total");
            recordsSuccessCounter = metricGroup.counter("records_success_total");
            recordsErrorCounter = metricGroup.counter("records_error_total");
            bytesProcessedCounter = metricGroup.counter("bytes_processed_total");
            
            // 创建Gauge指标
            processingRateGauge = metricGroup.gauge("processing_rate_per_second", () -> {
                long total = totalRecordsProcessed.get();
                if (jobStartTime != null) {
                    long seconds = Duration.between(jobStartTime, LocalDateTime.now()).getSeconds();
                    return seconds > 0 ? total / seconds : 0L;
                }
                return 0L;
            });
            
            successRateGauge = metricGroup.gauge("success_rate_percent", () -> {
                long total = totalRecordsProcessed.get();
                long success = totalRecordsSuccess.get();
                return total > 0 ? (double) success / total * 100.0 : 0.0;
            });
            
            errorRateGauge = metricGroup.gauge("error_rate_percent", () -> {
                long total = totalRecordsProcessed.get();
                long error = totalRecordsError.get();
                return total > 0 ? (double) error / total * 100.0 : 0.0;
            });
            
            log.info("Flink Metric监控系统初始化完成");
            
        } catch (Exception e) {
            log.warn("初始化Flink Metric监控失败: {}", e.getMessage());
        }
    }
    
    /**
     * 记录作业开始
     */
    public static void recordJobStart() {
        jobStartTime = LocalDateTime.now();
        lastProcessTime = jobStartTime;
        log.info("=== {} 开始执行 ===", jobName);
        log.info("作业开始时间: {}", formatTime(jobStartTime));
    }
    
    /**
     * 记录作业结束
     */
    public static void recordJobEnd() {
        jobEndTime = LocalDateTime.now();
        log.info("=== {} 执行完成 ===", jobName);
        log.info("作业结束时间: {}", formatTime(jobEndTime));
        printFinalStatistics();
    }
    
    /**
     * 记录处理成功的记录
     */
    public static void recordSuccess() {
        totalRecordsSuccess.incrementAndGet();
        totalRecordsProcessed.incrementAndGet();
        lastProcessTime = LocalDateTime.now();
        
        // 更新Flink Metric
        if (recordsSuccessCounter != null) {
            recordsSuccessCounter.inc();
        }
        if (recordsProcessedCounter != null) {
            recordsProcessedCounter.inc();
        }
    }
    
    /**
     * 记录处理失败的记录
     */
    public static void recordError() {
        totalRecordsError.incrementAndGet();
        totalRecordsProcessed.incrementAndGet();
        lastProcessTime = LocalDateTime.now();
        
        // 更新Flink Metric
        if (recordsErrorCounter != null) {
            recordsErrorCounter.inc();
        }
        if (recordsProcessedCounter != null) {
            recordsProcessedCounter.inc();
        }
    }
    
    /**
     * 记录处理的字节数
     * 
     * @param bytes 字节数
     */
    public static void recordBytesProcessed(long bytes) {
        totalBytesProcessed.addAndGet(bytes);
        
        // 更新Flink Metric
        if (bytesProcessedCounter != null) {
            bytesProcessedCounter.inc(bytes);
        }
    }
    
    /**
     * 记录处理记录（通用方法）
     * 
     * @param success 是否成功
     * @param bytes 字节数（可选）
     */
    public static void recordProcessed(boolean success, long bytes) {
        if (success) {
            recordSuccess();
        } else {
            recordError();
        }
        
        if (bytes > 0) {
            recordBytesProcessed(bytes);
        }
    }
    
    /**
     * 打印统计信息
     */
    public static void printStatistics() {
        long total = totalRecordsProcessed.get();
        long success = totalRecordsSuccess.get();
        long error = totalRecordsError.get();
        long bytes = totalBytesProcessed.get();
        
        double successRate = total > 0 ? (double) success / total * 100.0 : 0.0;
        double errorRate = total > 0 ? (double) error / total * 100.0 : 0.0;
        
        log.info("=== {} 统计信息 ===", jobName);
        log.info("总处理记录数: {}", total);
        log.info("  - 成功记录数: {} ({:.2f}%)", success, successRate);
        log.info("  - 失败记录数: {} ({:.2f}%)", error, errorRate);
        log.info("总处理字节数: {} bytes", bytes);
        log.info("最后处理时间: {}", formatTime(lastProcessTime));
        
        // 计算处理速率
        if (jobStartTime != null && lastProcessTime != null) {
            Duration duration = Duration.between(jobStartTime, lastProcessTime);
            long seconds = duration.getSeconds();
            if (seconds > 0) {
                double recordsPerSecond = (double) total / seconds;
                log.info("平均处理速率: {:.2f} 记录/秒", recordsPerSecond);
            }
        }
    }
    
    /**
     * 打印最终统计信息
     */
    public static void printFinalStatistics() {
        long total = totalRecordsProcessed.get();
        long success = totalRecordsSuccess.get();
        long error = totalRecordsError.get();
        long bytes = totalBytesProcessed.get();
        
        double successRate = total > 0 ? (double) success / total * 100.0 : 0.0;
        double errorRate = total > 0 ? (double) error / total * 100.0 : 0.0;
        
        log.info("=== {} 最终统计报告 ===", jobName);
        log.info("总处理记录数: {}", total);
        log.info("  - 成功记录数: {} ({:.2f}%)", success, successRate);
        log.info("  - 失败记录数: {} ({:.2f}%)", error, errorRate);
        log.info("总处理字节数: {} bytes", bytes);
        log.info("最后处理时间: {}", formatTime(lastProcessTime));
        
        // 计算总执行时间
        if (jobStartTime != null && jobEndTime != null) {
            Duration totalDuration = Duration.between(jobStartTime, jobEndTime);
            log.info("总执行时间: {} 秒", totalDuration.getSeconds());
            
            if (totalDuration.getSeconds() > 0) {
                double recordsPerSecond = (double) total / totalDuration.getSeconds();
                log.info("平均处理速率: {:.2f} 记录/秒", recordsPerSecond);
            }
        }
        
        log.info("===================================");
    }
    
    /**
     * 获取当前统计信息
     */
    public static JobStatistics getCurrentStatistics() {
        return new JobStatistics(
            jobName,
            totalRecordsProcessed.get(),
            totalRecordsSuccess.get(),
            totalRecordsError.get(),
            totalBytesProcessed.get(),
            jobStartTime,
            lastProcessTime
        );
    }
    
    /**
     * 格式化时间
     */
    private static String formatTime(LocalDateTime time) {
        return time != null ? time.format(TIME_FORMATTER) : "未知";
    }
    
    /**
     * 作业统计信息类
     */
    public static class JobStatistics {
        private final String jobName;
        private final long totalProcessed;
        private final long totalSuccess;
        private final long totalError;
        private final long totalBytes;
        private final LocalDateTime startTime;
        private final LocalDateTime lastProcessTime;
        
        public JobStatistics(String jobName, long totalProcessed, long totalSuccess, 
                           long totalError, long totalBytes, LocalDateTime startTime, 
                           LocalDateTime lastProcessTime) {
            this.jobName = jobName;
            this.totalProcessed = totalProcessed;
            this.totalSuccess = totalSuccess;
            this.totalError = totalError;
            this.totalBytes = totalBytes;
            this.startTime = startTime;
            this.lastProcessTime = lastProcessTime;
        }
        
        // Getters
        public String getJobName() { return jobName; }
        public long getTotalProcessed() { return totalProcessed; }
        public long getTotalSuccess() { return totalSuccess; }
        public long getTotalError() { return totalError; }
        public long getTotalBytes() { return totalBytes; }
        public LocalDateTime getStartTime() { return startTime; }
        public LocalDateTime getLastProcessTime() { return lastProcessTime; }
        
        public double getSuccessRate() {
            return totalProcessed > 0 ? (double) totalSuccess / totalProcessed * 100.0 : 0.0;
        }
        
        public double getErrorRate() {
            return totalProcessed > 0 ? (double) totalError / totalProcessed * 100.0 : 0.0;
        }
    }
}
